package com.webank.wedatasphere.linkis.metadata.service.impl;

import com.google.common.collect.Lists;
import com.webank.wedatasphere.linkis.metadata.domain.mdq.vo.MdqTablePartitionStatisticInfoVO;
import com.webank.wedatasphere.linkis.metadata.greenplum.dao.GpMetaDao;
import com.webank.wedatasphere.linkis.metadata.hive.config.DSEnum;
import com.webank.wedatasphere.linkis.metadata.hive.config.DataSource;
import com.webank.wedatasphere.linkis.metadata.service.DataSourceTypeService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @ClassName: DataSourceGpServiceImpl
 * @Description: 描述
 * @author: zys
 * @date: 2021/3/3  17:20
 */
@Service
public class DataSourceGpServiceImpl implements DataSourceTypeService {
    @Autowired
    GpMetaDao gpMetaDao;

    @DataSource(name = DSEnum.POSTGRESQL_DATA_SOURCE)
    @Override
    public List<Map<String, Object>> getTablesByDbNameAndUser(String database, Map<String, String> map) {
        List<Map<String, Object>> listTablesResult = Lists.newArrayList();
        List<Map<String, Object>> listTables = gpMetaDao.getTablesByDbNameAndUser(map);

        listTables.forEach(item -> {
            Map<String, Object> itemMap = new HashMap<>();
            item.keySet().forEach(key -> {
                itemMap.put(key.toUpperCase(), item.get(key));
            });
            listTablesResult.add(itemMap);
        });
        return listTablesResult;
    }

    @DataSource(name = DSEnum.POSTGRESQL_DATA_SOURCE)
    @Override
    public List<String> getDbsByUser(String userName) {
        List<String> dbs = gpMetaDao.getDbsByUser(userName);
        return dbs;
    }

    @DataSource(name = DSEnum.POSTGRESQL_DATA_SOURCE)
    @Override
    public List<Map<String, Object>> getColumns(String database, Map<String, String> param) {
        List<Map<String, Object>> columns = gpMetaDao.getColumns(param);
        //获取分区信息字段
        List<Map<String, Object>> partitionKeys = gpMetaDao.getPartitionKeys(param);
        Map<String, Boolean> keys = new HashMap<>();
        partitionKeys.forEach(partitionKey->{
            keys.put(partitionKey.get("columnname").toString(),true);
        });
        //遍历字段设置分区信息
        columns.forEach(col->{
            if(keys.containsKey(col.get("col_name"))){
                col.put("partitioned", true);
            }
        });
        return columns;
    }

    @DataSource(name = DSEnum.POSTGRESQL_DATA_SOURCE)
    @Override
    public List<Map<String, Object>> getPartitionKeys(Map<String, String> param) {
        List<Map<String, Object>> partitionKeys = gpMetaDao.getPartitionKeys(param);
        return partitionKeys;
    }

    @DataSource(name = DSEnum.POSTGRESQL_DATA_SOURCE)
    @Override
    public List<String> getPartitions(Map<String, String> param) {
        List<String> partitions = gpMetaDao.getPartitions(param);
        return partitions;
    }

    @Override
    public String getLocationByDbAndTable(Map<String, String> param) {
        return null;
    }

    @DataSource(name = DSEnum.POSTGRESQL_DATA_SOURCE)
    @Override
    public Map<String, Object> getTableDesc(Map<String, String> param) {
        SimpleDateFormat df =new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Map<String, Object> table = new HashMap<>();
        table.put("CREATE_TIME", "UNKNOWN");
        table.put("LAST_ACCESS_TIME", "UNKNOWN");
        table.put("OWNER", "");
        table.put("tab_name", param.get("tableName"));
        List<Map<String, Object>> tableDesc = gpMetaDao.getTableDesc(param);
        tableDesc.forEach(item->{
            try{
                Date createtime = df.parse(item.get("createtime").toString());
                String dfCreateDate = df.format(createtime);
                table.put("CREATE_TIME", dfCreateDate);
            }catch(Exception e){
                System.out.println("时间转换出错");
            }
            table.put("OWNER", item.get("owner"));
        });
        return table;
    }

    @Override
    public String getTableSize(Map<String,String> map) throws IOException {
        return null;
    }

    @Override
    public int getTableFileNum(Map<String,String> map) throws IOException {
        return 0;
    }

    @DataSource(name = DSEnum.POSTGRESQL_DATA_SOURCE)
    @Override
    public int getPartitionsNum(Map<String,String> map) throws IOException {
        List<String> partitions = gpMetaDao.getPartitions(map);
        return partitions.size();
    }

    @DataSource(name = DSEnum.POSTGRESQL_DATA_SOURCE)
    @Override
    public List<MdqTablePartitionStatisticInfoVO> getMdqTablePartitionStatisticInfoVO(Map<String,String> map,List<String> partitions, String tableLocation) throws IOException {
        List<MdqTablePartitionStatisticInfoVO> list = new ArrayList<>();
        partitions.sort((a, b) -> -a.compareTo(b));
        partitions.forEach(partition->{
            MdqTablePartitionStatisticInfoVO info=new MdqTablePartitionStatisticInfoVO();
            info.setName(partition);
            map.put("partitionName",partition);
            //当前只考虑了一个分区对应一个表的情况，并且没有使用子分区
            List<String> partitionTables = gpMetaDao.getPartitionTables(map);
            partitionTables.forEach(p->{
                String partitionTableSize = gpMetaDao.getPartitionTableSize(p);
                info.setPartitionSize(partitionTableSize);
            });
            list.add(info);
        });
        return list;
    }

    @DataSource(name = DSEnum.POSTGRESQL_DATA_SOURCE)
    @Override
    public int dropTable(Map<String, String> param) {
        int count = gpMetaDao.dropTable(param);
        return count;
    }
}
